package com.shujia.flink.table

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api._
import org.apache.flink.table.api.bridge.scala._
import org.apache.flink.types.Row

object DEmo2DSToTable {
  def main(args: Array[String]): Unit = {

    //创建flink 环境
    val bsEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    //设置table 环境的一些参数
    val bsSettings: EnvironmentSettings = EnvironmentSettings.newInstance()
      .useBlinkPlanner() //使用blikc计划器
      .inStreamingMode() //流模式
      .build()

    // 创建flink  table 环境
    val bsTableEnv: StreamTableEnvironment = StreamTableEnvironment.create(bsEnv, bsSettings)


    /**
      * 读取socket构建一个流
      *
      */

    val linesDS: DataStream[String] = bsEnv.socketTextStream("master", 8888)

    val studentDS: DataStream[(String, String, Int, String, String)] = linesDS.map(line => {
      val split: Array[String] = line.split(",")
      (split(0), split(1), split(2).toInt, split(3), split(4))
    })


    /**
      * 将流转换成一张表
      *
      */

    //在流上定义视图
    //bsTableEnv.createTemporaryView("student", studentDS, $"id", $"name", $"age", $"gender", $"clazz")

    //在流上定义表,Table 类似spark中的DataFrame
    val table: Table = bsTableEnv.fromDataStream(studentDS, $"id", $"name", $"age", $"gender", $"clazz")

    /**
      * DSL  在flink中使用不多，flink主流是使用sql
      *
      * 在动态表上进行连续查询得到一个新的动态表
      */
    val countTable: Table = table
      .groupBy($"clazz")
      .select($"clazz", $"clazz".count())

    /**
      * 将表转换成流
      * 1、仅追加的表 --  append流
      * 2、插入和更新的表  -- Retract流
      */


    val countDS: DataStream[(Boolean, Row)] = countTable.toRetractStream[Row]

    countDS.print()

    bsEnv.execute()

  }

}
